package rabbitmq

import (
	"context"
	"gitee.com/injoyai/goutil/oss/io"
	amqp "github.com/rabbitmq/amqp091-go"
)

var _ io.MessageReadCloser = new(Subscriber)

// Subscriber 订阅自动确认
type Subscriber struct {
	*Entity
	chanMessage chan []byte
}

func (this *Subscriber) ReadMessage() ([]byte, error) {
	return <-this.chanMessage, nil
}

func (this *Subscriber) Close() error {
	this.conn.Close()
	this.channel.Close()
	return nil
}

func NewSubscriber(cfg *Config) (*Subscriber, error) {
	return NewSubscriberWithContext(context.Background(), cfg)
}

func NewSubscriberWithContext(ctx context.Context, cfg *Config) (*Subscriber, error) {
	conn, err := amqp.Dial(cfg.Address)
	if err != nil {
		return nil, err
	}
	ch, err := conn.Channel()
	if err != nil {
		return nil, err
	}
	queue, err := ch.QueueDeclare(cfg.Name, cfg.Durable, cfg.AutoDelete, cfg.Exclusive, cfg.Nowait, nil)
	if err != nil {
		return nil, err
	}
	sub := &Subscriber{
		Entity: &Entity{
			cfg:     cfg,
			conn:    conn,
			queue:   queue,
			channel: ch,
			ctx:     ctx,
		},
		chanMessage: make(chan []byte),
	}
	msg, err := ch.Consume(queue.Name, "", true, false, false, false, nil)
	if err != nil {
		return nil, err
	}
	go func() {
		for v := range msg {
			if sub.cfg.Debug {
				Log.Writef("[订阅] %s", string(v.Body))
			}
			sub.chanMessage <- v.Body
		}
	}()
	return sub, nil
}
